package xxx;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.Writer;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

public class BlockingServer {

	private ServerSocket serverSocket;
	private ExecutorService threadPool;
	private AtomicBoolean running = new AtomicBoolean(false);
	
	public BlockingServer(int n) throws IOException {	
		threadPool = Executors.newFixedThreadPool(n);
		System.out.println("Server: Thread pool started " + threadPool);
	}
	
	public void start(int port) throws IOException {
		serverSocket = new ServerSocket();
		serverSocket.bind(new InetSocketAddress(port));
		System.out.println("Server: Listen " + serverSocket);
		try {
			running.set(true);
			while(running.get()) {
				final Socket socket = serverSocket.accept();
				System.out.println("Server: Accepted " + socket.toString());
				threadPool.submit(new Runnable() {
					public void run() {
						try {
							try {
								while(running.get() && !socket.isClosed()) {
									String line = read(socket);
									System.out.println("Server: Line " + line);
									write(socket, line);
									if("QUIT".equals(line)) {
										running.set(false);
										serverSocket.close();
										System.out.println("Server: Server socket closed " + serverSocket);
									}
								}
							} catch (EOFException e) {
								System.out.println("Server: EOF " + socket); 
							} finally {
								socket.close();
							}
						} catch (IOException e) {
							e.printStackTrace();
						}
					}});
				System.out.println("Server: Task submitted");	
			}
		} catch(SocketException e) {
			if(!"socket closed".equalsIgnoreCase(e.getMessage())) {
				throw e;
			}
		} finally {
			threadPool.shutdown();
			System.out.println("Server: Thread pool stoped " + threadPool);
			if(!serverSocket.isClosed()) {
				serverSocket.close();
				System.out.println("Server: Server socket closed " + serverSocket);
			}
		}
	}
	
	private String read(Socket socket) throws IOException {
		int length = readInt(socket);
		System.out.println("Server: Line length " + length);
		return readLine(length, socket);
	}
	
	private void write(Socket socket, String line) throws IOException {
		DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
		dos.writeInt(line.length());
		dos.flush();
		
		Writer w = new OutputStreamWriter(socket.getOutputStream());
		w.write(line);
		w.flush();
	}
	
	private int readInt(Socket socket) throws IOException {
		return new DataInputStream(socket.getInputStream()).readInt();
	}
	
	private String readLine(int length, Socket socket) throws IOException {
		char[] chars = new char[length];
		Reader r = new InputStreamReader(socket.getInputStream());
		r.read(chars);
		return String.valueOf(chars);
	}
	
	public static void main(String[] args) throws IOException {
		BlockingServer server = new BlockingServer(5);
		server.start(6667);
	}
	
}
